package com.sqllog.core;

import com.sqllog.config.CanalClientConfig;
import com.sqllog.listener.CanalEventListener;
import com.sqllog.listener.SimpleCanalEventListener;
import com.sqllog.listenerThread.MessageTransponder;
import com.sqllog.listenerThread.SimpleMessageTransponderRunnable;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

/**
 * 一个客户端可以存在多个连接
 *
 * @author chennengcheng
 * @date 2020-08-24 17:36
 * @remark
 */

@Slf4j
public class SimpleCanalClient extends AbstractCanalClient {

    /**
     *  监听处理器
     */
    private final List<CanalEventListener> listenerArrays = new ArrayList<>();

    /**
     * 转换工厂类
     */
    protected final TransponderRunnableFactory transponderRunnableFactory;

    /**
     * 消息转换线程
     */
    protected MessageTransponder messageTransponderRunner;

    /**
     * 根据配置初始化
     * @param canalConfig
     */
    public <T extends SimpleCanalEventListener> SimpleCanalClient(CanalClientConfig canalConfig, List<T> canalEventListener) {
        super(canalConfig);
        this.transponderRunnableFactory = SimpleMessageTransponderRunnable::new;
        listenerArrays.addAll(canalEventListener);
    }

    /**
     * 开启线程 开始连接
     * 这里不采用线程池是因为 线程是固定线程，会一直存在，且数量在启动时就已经确定
     * @param serverInstance
     */
    @Override
    protected void process(CanalClientConfig serverInstance, Thread.UncaughtExceptionHandler exceptionHandler){
        this.messageTransponderRunner = this.transponderRunnableFactory.newTransponder(serverInstance, this.listenerArrays);
        Thread thread = new Thread(this.messageTransponderRunner);
        thread.setUncaughtExceptionHandler(exceptionHandler);
        thread.start();
    }

    /**
     * 关闭 canal 客户端
     *
     * @param
     * @return
     */
    @Override
    public void stop() {
        //停止 canal 客户端
        this.messageTransponderRunner.stopRunning();
    }

    @Override
    public boolean isRunning(){
        return this.messageTransponderRunner.isRunning();
    }

}
